package cn.itcast.tags.models.rmd

import cn.itcast.tags.models.{AbstractModel, ModelType}
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.{ALS, ALSModel}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{DoubleType, StringType}
import org.apache.spark.storage.StorageLevel

import scala.util.matching.Regex

/**
 * 标签模型开发：品牌偏好BP模型
 * 使用基于隐语义模型的ALS推荐算法,构建用户购物偏好模型(商品/品牌偏好模型)
 */
class BpModel extends AbstractModel("品牌偏好BP模型", ModelType.ML) {
  /*
    382 用户购物偏好
   */
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    val spark: SparkSession = businessDF.sparkSession
    import spark.implicits._

    /*
      root
       |-- global_user_id: string (nullable = true)
       |-- loc_url: string (nullable = true)
     */
    // businessDF.printSchema()
    // businessDF.show(5, truncate = false)

    // TODO: 1. 自定义函数，从url中提取出访问商品id
    val url_to_product: UserDefinedFunction = udf(
      (url: String) => {
        // 正则表达式
        val regex: Regex = "^.+\\/product\\/(\\d+)\\.html.+$".r
        // 正则匹配
        val optionMatch: Option[Regex.Match] = regex.findFirstMatchIn(url)
        // 获取匹配的值
        val productId = optionMatch match {
          case Some(matchValue) => matchValue.group(1)
          case None => null
        }
        // 返回productId
        productId
      }
    )

    // TODO: 2. 从url中计算商品id
    val ratingsDF: Dataset[Row] = businessDF
      .filter($"loc_url".isNotNull) // 获取loc_url不为null
      .select(
        $"global_user_id".as("userId"),
        url_to_product($"loc_url").as("productId")
      ).
      filter($"productId".isNotNull) // 过滤不为空的数据
      // 统计每个用户点击各个商品的次数
      .groupBy($"userId", $"productId")
      .count()
      .select(
        $"userId".cast(DoubleType),
        $"productId".cast(DoubleType),
        $"count".as("rating").cast(DoubleType)
      )
    // ratingsDF.printSchema()
    // ratingsDF.show(10, truncate = false)
    // 数据缓存
    ratingsDF.persist(StorageLevel.MEMORY_AND_DISK).count()

    // TODO: 3. 使用ALS算法训练模型（评分为隐式评分）
    val alsModel: ALSModel = new ALS()
      // 设置属性
      .setUserCol("userId")
      .setItemCol("productId")
      .setRatingCol("rating")
      // 设置算法参数
      .setImplicitPrefs(true) // 隐式评分
      .setRank(10) // 矩阵因子，rank秩的值
      .setMaxIter(10) // 最大迭代次数
      .setColdStartStrategy("drop") // 冷启动
      .setAlpha(1.0)
      .setRegParam(1.0)
      // 应用数据集，训练模型
      .fit(ratingsDF)
    ratingsDF.unpersist()

    // TODO: 4. 模型评估
    val evaluator: RegressionEvaluator = new RegressionEvaluator()
      .setLabelCol("rating")
      .setPredictionCol("prediction")
      .setMetricName("rmse")
    val rmse: Double = evaluator.evaluate(alsModel.transform(ratingsDF))
    // rmse = 1.0300179222180903
    println(s"rmse = $rmse")

    // TODO: 5. 使用推荐模型推荐：
    /*
      推荐有两种方式：
      方式一：给用户推荐商品：当用户登录APP网站时，直接给用户推荐商品（Top10，Top5）
      方式二：给商品推荐用户：当用户登录APP网站时，将某个商品对给不同的用户
     */
    // 5.1 给用户推荐商品: Top5
    val rmdItemsDF: DataFrame = alsModel.recommendForAllUsers(5)
    /*
      root
       |-- userId: integer (nullable = false)
       |-- recommendations: array (nullable = true)
       |    |-- element: struct (containsNull = true)
       |    |    |-- productId: integer (nullable = true)
       |    |    |-- rating: float (nullable = true)
     */
    // rmdItemsDF.printSchema()
    rmdItemsDF.show(10, truncate = false)
    // 5.2. 给物品推荐用户
    val rmdUsersDF: DataFrame = alsModel.recommendForAllItems(5)
    // rmdUsersDF.printSchema()
    rmdUsersDF.show(10, truncate = false)

    // TODO: 6. 存储推荐数据
    /*
      给用户推荐的商品存储到HBase表中：tbl_rmd_items，
      列簇：info, RowKey: userId, Column: items
     */
    val modelDF: DataFrame = rmdItemsDF
      .select(
        $"userId",
        $"recommendations.productId".as("productIds"),
        $"recommendations.rating".as("ratings")
      ).
      select(
        $"userId".cast(StringType),
        concat_ws(",", $"productIds").as("productIds"),
        concat_ws(",", $"ratings").as("ratings")
      )
    modelDF.printSchema()
    modelDF.show(10, truncate = false)

    // TODO: 7. 保存推荐的结果到HBase表中
    modelDF.write
      .mode(SaveMode.Overwrite)
      .format("hbase")
      .option("zkHosts", "bigdata-cdh01.itcast.cn")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_rmd_items")
      .option("family", "info")
      .option("rowKeyColumn", "userId")
      .save()
    null
  }
}

object BpModel {
  def main(args: Array[String]): Unit = {
    val tagModel = new BpModel()
    tagModel.executeModel(382L)
  }
}